package main

import (
	"context"
	"database/sql"
	"fmt"
	"os"
	"regexp"
	"strconv"
	"strings"
	"time"

	"kwdb-import/pkg/config"
	"kwdb-import/pkg/converter"
	"kwdb-import/pkg/errors"
	"kwdb-import/pkg/executor"
	"kwdb-import/pkg/logger"
	"kwdb-import/pkg/parser"

	_ "github.com/lib/pq"
	"github.com/spf13/cobra"
)

var (
	// 命令行参数
	connection string
	username   string
	password   string
	configFile string
	logLevel   string
	logFile    string
	batchSize  int
	timeout    string
	maxRetries int
	outputDir  string
	verbose    bool
	dryRun     bool
	version    = "1.3.0"

	// 全局配置和日志
	globalConfig *config.Config
	globalLogger logger.Logger
)

var rootCmd = &cobra.Command{
	Use:   "kwdb-import [SQL文件路径]",
	Short: "KWDB异构数据库快速迁移导入工具",
	Long: `KWDB异构数据库快速迁移导入工具是一个专门用于将MySQL数据库结构和数据迁移到KWDB关系引擎的命令行工具。
该工具能够解析MySQL的SQL转储文件，并将其转换为KWDB兼容的SQL语句。

支持功能:
- MySQL到KWDB的SQL语句转换
- 数据类型自动映射
- 批量导入优化
- 事务管理
- 错误处理和重试
- 详细的日志记录
- 配置文件支持`,
	Args: cobra.MaximumNArgs(1),
	PersistentPreRunE: func(cmd *cobra.Command, args []string) error {
		return initializeApplication()
	},
	RunE: func(cmd *cobra.Command, args []string) error {
		return runImport(args)
	},
}

var versionCmd = &cobra.Command{
	Use:   "version",
	Short: "显示版本信息",
	Run: func(cmd *cobra.Command, args []string) {
		fmt.Printf("kwdb-import version %s\n", version)
	},
}

var validateCmd = &cobra.Command{
	Use:   "validate [配置文件路径]",
	Short: "验证配置文件",
	Long:  "验证配置文件的格式和内容是否正确，并显示验证结果和建议",
	Args:  cobra.MaximumNArgs(1),
	RunE: func(cmd *cobra.Command, args []string) error {
		var configPath string
		if len(args) > 0 {
			configPath = args[0]
		} else if configFile != "" {
			configPath = configFile
		} else {
			// 未显式指定时，尝试默认文件名
			candidates := []string{"config.yaml", "config.yml", "config.json"}
			for _, cand := range candidates {
				if _, err := os.Stat(cand); err == nil {
					configPath = cand
					break
				}
			}
			if configPath == "" {
				return fmt.Errorf("必须指定配置文件路径，或在当前目录提供 config.yaml/config.yml/config.json")
			}
		}

		// 加载配置
		cfg, err := config.LoadFromFile(configPath)
		if err != nil {
			return fmt.Errorf("加载配置文件失败: %v", err)
		}

		// 验证配置
		if err := cfg.Validate(); err != nil {
			fmt.Printf("❌ 配置验证失败:\n%v\n", err)
			return err
		}

		fmt.Printf("✅ 配置验证通过\n")

		// 显示配置摘要
		fmt.Printf("\n配置摘要:\n")
		// 使用顶层 DBName（已与嵌套 Database 同步）
		fmt.Printf("  数据库: %s@%s:%d/%s\n", cfg.Username, cfg.Host, cfg.Port, cfg.DBName)
		fmt.Printf("  批处理大小: %d\n", cfg.BatchSize)
		fmt.Printf("  超时时间: %s\n", cfg.Timeout)
		fmt.Printf("  日志级别: %s\n", cfg.LogLevel)
		fmt.Printf("  数据类型映射数量: %d\n", len(cfg.DataTypeMapping))
		if len(cfg.DataTypeMapping) > 0 {
			fmt.Printf("\n数据类型映射:\n")
			for k, v := range cfg.DataTypeMapping {
				fmt.Printf("  %s -> %s\n", k, v)
			}
		}

		// 显示建议
		suggestions := cfg.GetSuggestions()
		if len(suggestions) > 0 {
			fmt.Printf("\n💡 配置建议:\n")
			for _, suggestion := range suggestions {
				fmt.Printf("  - %s\n", suggestion)
			}
		}

		return nil
	},
}

var configCmd = &cobra.Command{
	Use:   "config",
	Short: "配置管理",
	Long:  "配置文件的创建、查看和管理功能",
}

var configInitCmd = &cobra.Command{
	Use:   "init [配置文件路径]",
	Short: "创建默认配置文件",
	Long:  "创建一个包含默认设置的配置文件模板",
	Args:  cobra.MaximumNArgs(1),
	RunE: func(cmd *cobra.Command, args []string) error {
		configPath := "config.yaml"
		if len(args) > 0 {
			configPath = args[0]
		}

		// 检查文件是否已存在
		if _, err := os.Stat(configPath); err == nil {
			return fmt.Errorf("配置文件已存在: %s", configPath)
		}

		// 创建默认配置
		cfg := config.NewConfig()

		// 保存到文件
		if err := cfg.SaveToFile(configPath); err != nil {
			return fmt.Errorf("保存配置文件失败: %v", err)
		}

		fmt.Printf("✅ 默认配置文件已创建: %s\n", configPath)
		fmt.Printf("\n请编辑配置文件并设置以下必需参数:\n")
		fmt.Printf("  - host: 数据库主机地址\n")
		fmt.Printf("  - port: 数据库端口\n")
		fmt.Printf("  - database: 数据库名称\n")
		fmt.Printf("  - username: 数据库用户名\n")
		fmt.Printf("  - password: 数据库密码\n")

		return nil
	},
}

var configShowCmd = &cobra.Command{
	Use:   "show [配置文件路径]",
	Short: "显示当前配置",
	Long:  "显示当前配置文件的内容和有效配置",
	Args:  cobra.MaximumNArgs(1),
	RunE: func(cmd *cobra.Command, args []string) error {
		var configPath string
		if len(args) > 0 {
			configPath = args[0]
		} else if configFile != "" {
			configPath = configFile
		} else {
			// 未显式指定时，尝试默认文件名
			candidates := []string{"config.yaml", "config.yml", "config.json"}
			for _, cand := range candidates {
				if _, err := os.Stat(cand); err == nil {
					configPath = cand
					break
				}
			}
			if configPath == "" {
				return fmt.Errorf("必须指定配置文件路径，或在当前目录提供 config.yaml/config.yml/config.json")
			}
		}

		// 加载配置
		cfg, err := config.LoadFromFile(configPath)
		if err != nil {
			return fmt.Errorf("加载配置文件失败: %v", err)
		}

		// 显示配置
		fmt.Printf("配置文件: %s\n\n", configPath)
		fmt.Printf("数据库配置:\n")
		fmt.Printf("  主机: %s\n", cfg.Host)
		fmt.Printf("  端口: %d\n", cfg.Port)
		// 顶层 DBName 与嵌套 Database 已在加载时同步
		fmt.Printf("  数据库: %s\n", cfg.DBName)
		fmt.Printf("  用户名: %s\n", cfg.Username)
		fmt.Printf("  密码: %s\n", maskPassword(cfg.Password))

		fmt.Printf("\n导入配置:\n")
		fmt.Printf("  批处理大小: %d\n", cfg.BatchSize)
		fmt.Printf("  超时时间: %s\n", cfg.Timeout)
		fmt.Printf("  最大重试次数: %d\n", cfg.MaxRetries)
		fmt.Printf("  最大错误数: %d\n", cfg.MaxErrors)
		fmt.Printf("  错误处理策略: %s\n", cfg.OnError)

		fmt.Printf("\n日志配置:\n")
		fmt.Printf("  日志级别: %s\n", cfg.LogLevel)
		fmt.Printf("  日志格式: %s\n", cfg.LogFormat)
		fmt.Printf("  日志文件: %s\n", cfg.LogFile)

		fmt.Printf("\n性能配置:\n")
		fmt.Printf("  并发启用: %v\n", cfg.EnableParallel)
		fmt.Printf("  并发工作协程: %d\n", cfg.WorkerCount)
		fmt.Printf("  并发批次大小: %d\n", cfg.ParallelBatchSize)
		fmt.Printf("  压缩启用: %v\n", cfg.EnableCompression)
		fmt.Printf("  内存限制(MB): %d\n", cfg.MemoryLimit)

		return nil
	},
}

// dbstatsCmd 展示目标数据库表数量、每张表名及总行数、视图数量、视图名称
var dbstatsCmd = &cobra.Command{
	Use:   "dbstats",
	Short: "展示目标数据库表/视图统计信息",
	Long:  "展示目标数据库表数量、每张表名及总行数（默认估算，可选精确），以及视图数量与名称。",
	Args:  cobra.MaximumNArgs(0),
	RunE: func(cmd *cobra.Command, args []string) error {
		if err := initializeApplication(); err != nil {
			return err
		}

		// 读取标志
		schema, _ := cmd.Flags().GetString("schema")
		exact, _ := cmd.Flags().GetBool("exact")

		// 建立数据库连接
		connStr := globalConfig.GetConnectionString()
		db, err := sqlOpenPostgres(connStr)
		if err != nil {
			return fmt.Errorf("连接数据库失败: %v", err)
		}
		defer db.Close()

		// 使用查询超时
		timeout := globalConfig.QueryTimeout
		if timeout == 0 {
			timeout = 30 * time.Second
		}
		ctx, cancel := context.WithTimeout(context.Background(), timeout)
		defer cancel()

		// 查询表列表
		var tableRows *RowsResult[string]
		if schema == "" {
			tableRows, err = queryStrings(ctx, db, `SELECT table_schema||'.'||table_name AS ident
                FROM information_schema.tables
                WHERE table_type='BASE TABLE' AND table_schema NOT IN ('pg_catalog','information_schema')
                ORDER BY table_schema, table_name`)
		} else {
			tableRows, err = queryStrings(ctx, db, `SELECT table_schema||'.'||table_name AS ident
                FROM information_schema.tables
                WHERE table_type='BASE TABLE' AND table_schema = $1
                ORDER BY table_schema, table_name`, schema)
		}
		if err != nil {
			return fmt.Errorf("查询表列表失败: %v", err)
		}

		// 查询视图列表
		var viewRows *RowsResult[string]
		if schema == "" {
			viewRows, err = queryStrings(ctx, db, `SELECT table_schema||'.'||table_name AS ident
                FROM information_schema.views
                WHERE table_schema NOT IN ('pg_catalog','information_schema')
                ORDER BY table_schema, table_name`)
		} else {
			viewRows, err = queryStrings(ctx, db, `SELECT table_schema||'.'||table_name AS ident
                FROM information_schema.views
                WHERE table_schema = $1
                ORDER BY table_schema, table_name`, schema)
		}
		if err != nil {
			return fmt.Errorf("查询视图列表失败: %v", err)
		}

		// 计算行数
		rowCounts := make(map[string]int64)
		if exact {
			for _, ident := range tableRows.Items {
				parts := strings.SplitN(ident, ".", 2)
				if len(parts) != 2 { // 防御
					continue
				}
				count, cerr := queryCountExact(ctx, db, parts[0], parts[1])
				if cerr != nil {
					globalLogger.Warn("获取精确行数失败，使用估算",
						logger.String("table", ident),
						logger.ErrorField(cerr),
					)
					// 尝试估算
					est, eerr := queryCountEstimate(ctx, db, parts[0], parts[1])
					if eerr == nil {
						rowCounts[ident] = est
					}
					continue
				}
				rowCounts[ident] = count
			}
		} else {
			// 批量估算
			var estRows *RowsKV[string, int64]
			if schema == "" {
				estRows, err = queryEstimateAll(ctx, db)
			} else {
				estRows, err = queryEstimateSchema(ctx, db, schema)
			}
			if err != nil {
				globalLogger.Warn("批量估算失败，行数将显示为0或未知", logger.ErrorField(err))
			} else {
				for k, v := range estRows.Map {
					rowCounts[k] = v
				}
			}

			// 如果估算结果全部为0或缺失，自动回退到精确统计（确保输出不误导）
			nonZero := 0
			for _, ident := range tableRows.Items {
				if rowCounts[ident] > 0 {
					nonZero++
				}
			}
			if nonZero == 0 {
				globalLogger.Info("估算不可用或为0，回退到精确 COUNT(*) 以获取真实行数")
				for _, ident := range tableRows.Items {
					parts := strings.SplitN(ident, ".", 2)
					if len(parts) != 2 {
						continue
					}
					c, cerr := queryCountExact(ctx, db, parts[0], parts[1])
					if cerr == nil {
						rowCounts[ident] = c
					} else {
						// 最后尝试单表估算，若也失败则保持0
						if est, eerr := queryCountEstimate(ctx, db, parts[0], parts[1]); eerr == nil {
							rowCounts[ident] = est
						}
					}
				}
			}
		}

		// 输出
		fmt.Println("=== 数据库对象统计 ===")
		fmt.Printf("表数量: %d\n", len(tableRows.Items))
		for _, ident := range tableRows.Items {
			fmt.Printf("- %s 行数: %d\n", ident, rowCounts[ident])
		}
		fmt.Printf("视图数量: %d\n", len(viewRows.Items))
		for _, ident := range viewRows.Items {
			fmt.Printf("- %s\n", ident)
		}

		return nil
	},
}

// sqlOpenPostgres 打开数据库连接（lib/pq）
func sqlOpenPostgres(conn string) (*sql.DB, error) {
	return sql.Open("postgres", conn)
}

// RowsResult 简单结果封装
type RowsResult[T any] struct {
	Items []T
}

// RowsKV 键值结果封装
type RowsKV[K comparable, V any] struct {
	Map map[K]V
}

// queryStrings 查询单列字符串
func queryStrings(ctx context.Context, db *sql.DB, sqlText string, args ...any) (*RowsResult[string], error) {
	rows, err := db.QueryContext(ctx, sqlText, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	res := &RowsResult[string]{Items: make([]string, 0)}
	for rows.Next() {
		var s string
		if err := rows.Scan(&s); err != nil {
			return nil, err
		}
		res.Items = append(res.Items, s)
	}
	return res, rows.Err()
}

// queryCountExact 对单表做精确 COUNT(*)
func queryCountExact(ctx context.Context, db *sql.DB, schema string, table string) (int64, error) {
	ident := quoteIdent(schema) + "." + quoteIdent(table)
	q := "SELECT COUNT(*) FROM " + ident
	rows, err := db.QueryContext(ctx, q)
	if err != nil {
		return 0, err
	}
	defer rows.Close()
	var c int64
	if rows.Next() {
		if err := rows.Scan(&c); err != nil {
			return 0, err
		}
	}
	return c, rows.Err()
}

// queryCountEstimate 从pg_stat_all_tables获取估算行数
func queryCountEstimate(ctx context.Context, db *sql.DB, schema string, table string) (int64, error) {
	rows, err := db.QueryContext(ctx, `SELECT n_live_tup FROM pg_stat_all_tables WHERE schemaname=$1 AND relname=$2`, schema, table)
	if err != nil {
		return 0, err
	}
	defer rows.Close()
	var c int64
	if rows.Next() {
		if err := rows.Scan(&c); err != nil {
			return 0, err
		}
	}
	return c, rows.Err()
}

// queryEstimateAll 批量获取所有用户表估算行数
func queryEstimateAll(ctx context.Context, db *sql.DB) (*RowsKV[string, int64], error) {
	rows, err := db.QueryContext(ctx, `SELECT schemaname||'.'||relname AS ident, n_live_tup FROM pg_stat_all_tables WHERE schemaname NOT IN ('pg_catalog','information_schema')`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	res := &RowsKV[string, int64]{Map: make(map[string]int64)}
	for rows.Next() {
		var ident string
		var n int64
		if err := rows.Scan(&ident, &n); err != nil {
			return nil, err
		}
		res.Map[ident] = n
	}
	return res, rows.Err()
}

// queryEstimateSchema 批量获取某schema估算行数
func queryEstimateSchema(ctx context.Context, db *sql.DB, schema string) (*RowsKV[string, int64], error) {
	rows, err := db.QueryContext(ctx, `SELECT schemaname||'.'||relname AS ident, n_live_tup FROM pg_stat_all_tables WHERE schemaname=$1`, schema)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	res := &RowsKV[string, int64]{Map: make(map[string]int64)}
	for rows.Next() {
		var ident string
		var n int64
		if err := rows.Scan(&ident, &n); err != nil {
			return nil, err
		}
		res.Map[ident] = n
	}
	return res, rows.Err()
}

// quoteIdent SQL标识符双引号转义
func quoteIdent(s string) string {
	return "\"" + strings.ReplaceAll(s, "\"", "\"\"") + "\""
}

// maskPassword 遮蔽密码显示
func maskPassword(password string) string {
	if password == "" {
		return "(未设置)"
	}
	if len(password) <= 2 {
		return "***"
	}
	return password[:2] + strings.Repeat("*", len(password)-2)
}

func init() {
	// 数据库连接参数
	rootCmd.PersistentFlags().StringVarP(&connection, "connection", "c", "", "数据库连接字符串 (格式: host:port/database)")
	rootCmd.PersistentFlags().StringVarP(&username, "username", "u", "", "数据库用户名")
	rootCmd.PersistentFlags().StringVarP(&password, "password", "p", "", "数据库密码")

	// 配置文件参数
	rootCmd.PersistentFlags().StringVar(&configFile, "config", "", "配置文件路径 (支持YAML/JSON格式)")

	// 日志参数
	rootCmd.PersistentFlags().StringVar(&logLevel, "log-level", "info", "日志级别 (debug, info, warn, error, fatal)")
	rootCmd.PersistentFlags().StringVar(&logFile, "log-file", "", "日志文件路径 (默认输出到控制台)")
	rootCmd.PersistentFlags().BoolVarP(&verbose, "verbose", "v", false, "详细输出模式")

	// 导入参数
	rootCmd.PersistentFlags().IntVarP(&batchSize, "batch-size", "b", 1000, "批处理大小")
	rootCmd.PersistentFlags().StringVarP(&timeout, "timeout", "t", "30s", "操作超时时间")
	rootCmd.PersistentFlags().IntVar(&maxRetries, "max-retries", 3, "最大重试次数")
	rootCmd.PersistentFlags().StringVarP(&outputDir, "output-dir", "o", "", "输出目录 (用于日志和错误文件)")
	rootCmd.PersistentFlags().BoolVar(&dryRun, "dry-run", false, "试运行模式 (仅解析和转换，不执行SQL)")

	// 添加子命令
	rootCmd.AddCommand(versionCmd)
	rootCmd.AddCommand(validateCmd)
	rootCmd.AddCommand(configCmd)
	rootCmd.AddCommand(dbstatsCmd)

	// 添加config子命令
	configCmd.AddCommand(configInitCmd)
	configCmd.AddCommand(configShowCmd)

	// dbstats 选项
	dbstatsCmd.Flags().String("schema", "", "限定schema（默认统计所有用户schema）")
	dbstatsCmd.Flags().Bool("exact", false, "使用精确COUNT(*)（可能较慢），默认使用估算n_live_tup")
}

// initializeApplication 初始化应用程序
func initializeApplication() error {
	// 初始化配置
	var err error
	if configFile != "" {
		// 从配置文件加载
		globalConfig, err = config.LoadFromFile(configFile)
		if err != nil {
			return fmt.Errorf("加载配置文件失败: %v", err)
		}
	} else {
		// 未指定 --config 时，尝试自动加载当前目录下的默认配置文件
		candidates := []string{"config.yaml", "config.yml", "config.json"}
		for _, cand := range candidates {
			if _, statErr := os.Stat(cand); statErr == nil {
				loaded, loadErr := config.LoadFromFile(cand)
				if loadErr != nil {
					return fmt.Errorf("加载默认配置文件失败(%s): %v", cand, loadErr)
				}
				globalConfig = loaded
				// 记录选用的配置文件，便于日志显示
				configFile = cand
				break
			}
		}
		if globalConfig == nil {
			// 没有找到默认配置文件，退回到内置默认配置
			globalConfig = config.NewConfig()
		}
	}

	// 应用命令行参数覆盖
	applyCommandLineOverrides()

	// 验证配置
	if err := globalConfig.Validate(); err != nil {
		return fmt.Errorf("配置验证失败: %v", err)
	}

	// 初始化日志
	loggerConfig := &logger.Config{
		Level:  parseLogLevel(globalConfig.LogLevel),
		Format: logger.TextFormat,
		Output: globalConfig.LogFile,
		Caller: verbose,
	}

	if globalConfig.LogFormat == "json" {
		loggerConfig.Format = logger.JSONFormat
	}

	globalLogger, err = logger.NewLogger(loggerConfig)
	if err != nil {
		return fmt.Errorf("初始化日志失败: %v", err)
	}

	// 设置全局日志
	logger.SetGlobalLogger(globalLogger)

	globalLogger.Info("应用程序初始化完成",
		logger.String("version", version),
		logger.String("config_file", configFile),
		logger.String("log_level", globalConfig.LogLevel),
	)

	return nil
}

// applyCommandLineOverrides 应用命令行参数覆盖
func applyCommandLineOverrides() {
	if connection != "" {
		// 解析连接字符串
		parts := strings.Split(connection, "/")
		if len(parts) == 2 {
			hostPort := parts[0]
			globalConfig.DBName = parts[1]

			hostPortParts := strings.Split(hostPort, ":")
			if len(hostPortParts) == 2 {
				globalConfig.Host = hostPortParts[0]
				if port := hostPortParts[1]; port != "" {
					// 更稳健的端口解析
					if p, err := strconv.Atoi(port); err == nil {
						globalConfig.Port = p
					}
				}
			}
		}
	}

	if username != "" {
		globalConfig.Username = username
	}
	if password != "" {
		globalConfig.Password = password
	}
	if logLevel != "" {
		globalConfig.LogLevel = logLevel
	}
	if logFile != "" {
		globalConfig.LogFile = logFile
	}
	if batchSize > 0 {
		globalConfig.BatchSize = batchSize
	}
	if timeout != "" {
		globalConfig.Timeout = timeout
	}
	if maxRetries > 0 {
		globalConfig.MaxRetries = maxRetries
	}
	if outputDir != "" {
		globalConfig.OutputDir = outputDir
	}
	if verbose {
		globalConfig.LogLevel = "debug"
	}
	// 将命令行的 dry-run 映射到全局配置，确保两者一致
	if dryRun {
		globalConfig.DryRun = true
	}
}

// parseLogLevel 解析日志级别
func parseLogLevel(level string) logger.LogLevel {
	parsedLevel, err := logger.ParseLogLevel(level)
	if err != nil {
		return logger.InfoLevel // 默认使用info级别
	}
	return parsedLevel
}

// runImport 执行导入操作
func runImport(args []string) error {
	// 确定输入文件
	var inputFile string
	if len(args) > 0 {
		inputFile = args[0]
	} else if globalConfig.InputFile != "" {
		inputFile = globalConfig.InputFile
	} else {
		return errors.NewValidationError("input_file cannot be empty", "input_file", inputFile)
	}

	// 验证文件路径
	fileValidator := config.NewValidator()
	fileValidator.AddRule("input_file", &config.FilePathRule{FieldName: "input_file", MustExist: true, MustBeFile: true})
	if err := fileValidator.ValidateField("input_file", inputFile); err != nil {
		return err
	}
	mapStr := fmt.Sprintf("%v", globalConfig.DataTypeMapping)
	globalLogger.Info("开始导入操作",
		logger.String("input_file", inputFile),
		logger.String("target_host", globalConfig.Host),
		logger.Int("target_port", globalConfig.Port),
		logger.String("target_database", globalConfig.DBName),
		logger.Int("batch_size", globalConfig.BatchSize),
		logger.Bool("dry_run", globalConfig.DryRun),
		logger.String("数据类型映射", mapStr),
	)

	// 创建解析器
	globalLogger.Debug("创建SQL解析器")
	sqlParser := parser.NewMySQLParser()

	// 流式执行分支：避免一次性加载所有语句到内存
	if strings.EqualFold(globalConfig.ExecutionMode, "streaming") {
		globalLogger.Info("启用流式执行模式以降低内存占用")

		// 创建转换器
		converterInstance := converter.NewMySQLToKWDBConverter()

		// 如果是试运行模式：流式逐条输出转换结果（保持原有顺序）
		if globalConfig.DryRun {
			globalLogger.Info("试运行流式模式，逐条输出转换结果")
			start := time.Now()
			convertedCount := 0
			err := sqlParser.ParseStream(inputFile, func(stmt parser.SQLStatement) error {
				converted, err := converterInstance.Convert(stmt, globalConfig.DataTypeMapping)
				if err != nil {
					return errors.WrapError(err, errors.ConversionError, "SQL语句转换失败")
				}
				if strings.TrimSpace(converted) == "" {
					return nil
				}
				// 对转换结果进行顶层分号拆分，逐条输出
				parts := parser.SplitTopLevelStatements(converted)
				if len(parts) == 0 {
					return nil
				}
				for idx, part := range parts {
					p := strings.TrimSpace(part)
					if p == "" {
						continue
					}
					// 按片段重新识别类型，避免误标
					partType := parser.ClassifyStatement(p)
					stmtInfo := fmt.Sprintf("-- Line %d (%s) split %d/%d\n%s", stmt.Line, parser.GetStatementTypeName(partType), idx+1, len(parts), p)
					fmt.Printf("%s\n\n", stmtInfo)
					globalLogger.Info("DRY-RUN转换结果",
						logger.Int("line", stmt.Line),
						logger.String("type", parser.GetStatementTypeName(partType)),
						logger.Int("split_index", idx+1),
						logger.Int("split_total", len(parts)),
						logger.String("original", strings.TrimSpace(stmt.Content)),
						logger.String("converted", p),
					)
					convertedCount++
				}
				return nil
			})
			if err != nil {
				return err
			}
			globalLogger.Info("试运行(流式)完成",
				logger.Int("converted_count", convertedCount),
				logger.Duration("total_time", time.Since(start)),
			)
			fmt.Printf("\n=== 试运行完成: 共转换 %d 条语句，耗时 %v ===\n", convertedCount, time.Since(start))
			return nil
		}

		// 创建执行器
		globalLogger.Debug("创建数据库执行器")
		exec, err := executor.NewPgxStatementExecutor(globalConfig.GetConnectionString())
		if err != nil {
			return errors.WrapError(err, errors.ConnectionError, "创建数据库执行器失败")
		}
		defer func() {
			if closeErr := exec.Close(); closeErr != nil {
				globalLogger.Error("关闭数据库连接失败", logger.ErrorField(closeErr))
			}
		}()

		// 测试数据库连接
		globalLogger.Info("测试数据库连接")
		if err := exec.ValidateConnection(); err != nil {
			return errors.WrapError(err, errors.ConnectionError, "数据库连接测试失败")
		}

		// 执行SQL语句（两阶段流式执行）
		globalLogger.Info("开始执行SQL语句(流式)")
		startTime := time.Now()
		executedCount := 0
		errorCount := 0

		// 第一阶段：非DML（DDL/视图/索引等）。遇到依赖缺失先延后。
		ddlRetry := make([]string, 0)
		err = sqlParser.ParseStream(inputFile, func(stmt parser.SQLStatement) error {
			// 跳过DML语句（INSERT/UPDATE/DELETE）在第二阶段处理
			origUp := strings.ToUpper(strings.TrimSpace(stmt.Content))
			if strings.HasPrefix(origUp, "INSERT ") || strings.HasPrefix(origUp, "INSERT INTO ") ||
				strings.HasPrefix(origUp, "UPDATE ") || strings.HasPrefix(origUp, "DELETE ") {
				return nil
			}

			// 基于原始分类跳过不兼容的过程/函数/触发器和DELIMITER
			if stmt.Type == parser.CreateProcedure || stmt.Type == parser.CreateFunction || stmt.Type == parser.CreateTrigger || stmt.Type == parser.Delimiter {
				globalLogger.Info("跳过不兼容的MySQL过程/函数或分隔符语句", logger.String("sql", strings.TrimSpace(stmt.Content)))
				return nil
			}

			converted, err := converterInstance.Convert(stmt, globalConfig.DataTypeMapping)
			if err != nil {
				return errors.WrapError(err, errors.ConversionError, "SQL语句转换失败")
			}
			// 对转换结果进行顶层分号拆分并逐条执行
			parts := parser.SplitTopLevelStatements(strings.TrimSpace(converted))
			for _, part := range parts {
				trimmed := strings.TrimSpace(part)
				if trimmed == "" {
					continue
				}

				// 跳过 psql 元命令（例如 \c / \connect）
				trimNoSpace := strings.TrimSpace(trimmed)
				if strings.HasPrefix(trimNoSpace, "\\c ") || strings.HasPrefix(trimNoSpace, "\\connect ") {
					globalLogger.Info("跳过psql元命令", logger.String("sql", trimmed))
					continue
				}

				up := strings.ToUpper(trimmed)
				// 忽略执行 CREATE/DROP DATABASE/USE
				if strings.HasPrefix(up, "CREATE DATABASE") || strings.HasPrefix(up, "DROP DATABASE") || strings.HasPrefix(up, "USE ") {
					globalLogger.Info("跳过数据库级语句执行", logger.String("sql", trimmed))
					continue
				}
				// 忽略MySQL过程/函数/DELIMITER相关
				if strings.HasPrefix(up, "CREATE PROCEDURE") || strings.HasPrefix(up, "CREATE FUNCTION") ||
					strings.HasPrefix(up, "DROP PROCEDURE") || strings.HasPrefix(up, "DROP FUNCTION") ||
					strings.HasPrefix(up, "DELIMITER ") || strings.HasPrefix(up, "DELIMITER\t") {
					globalLogger.Info("跳过不兼容的MySQL过程/函数或分隔符语句", logger.String("sql", trimmed))
					continue
				}
				// 跳过MySQL存储过程结束标记
				reEnd := regexp.MustCompile(`(?i)^\s*END\s*(;{1,2}|//|\$\$)?\s*$`)
				if reEnd.MatchString(trimmed) {
					globalLogger.Debug("跳过MySQL存储过程结束标记", logger.String("sql", trimmed))
					continue
				}
				// 跳过锁表
				if strings.HasPrefix(up, "LOCK TABLES") || strings.HasPrefix(up, "UNLOCK TABLES") {
					globalLogger.Debug("跳过MySQL表级锁语句", logger.String("sql", trimmed))
					continue
				}
				// 跳过MySQL的ALTER TABLE ENABLE/DISABLE KEYS（通常在/*!版本化注释*/中）
				if strings.Contains(up, "ALTER TABLE") && (strings.Contains(up, "ENABLE KEYS") || strings.Contains(up, "DISABLE KEYS")) {
					globalLogger.Debug("跳过MySQL ALTER TABLE KEYS语句", logger.String("sql", trimmed))
					continue
				}
				// 跳过事务控制
				if strings.HasPrefix(up, "BEGIN") || strings.HasPrefix(up, "START TRANSACTION") ||
					strings.HasPrefix(up, "COMMIT") || strings.HasPrefix(up, "ROLLBACK") ||
					strings.HasPrefix(up, "SAVEPOINT") || strings.HasPrefix(up, "RELEASE SAVEPOINT") ||
					strings.HasPrefix(up, "SET AUTOCOMMIT") {
					globalLogger.Debug("跳过事务控制语句", logger.String("sql", trimmed))
					continue
				}

				if err := exec.ExecuteSingleStatement(trimmed); err != nil {
					if isRelationMissing(err) && ((strings.HasPrefix(up, "CREATE TABLE") && strings.Contains(up, "FOREIGN KEY")) || strings.HasPrefix(up, "CREATE VIEW") || strings.HasPrefix(up, "CREATE MATERIALIZED VIEW")) {
						ddlRetry = append(ddlRetry, trimmed)
						globalLogger.Warn("检测到DDL依赖未创建，延后重试",
							logger.String("sql", trimmed),
							logger.ErrorField(err),
						)
						continue
					}
					errorCount++
					globalLogger.Error("SQL执行失败(非DML阶段)", logger.String("sql", trimmed), logger.ErrorField(err))
					if errorCount >= globalConfig.MaxErrors {
						return errors.NewExecutionError(fmt.Sprintf("错误数量超过限制 (%d)", globalConfig.MaxErrors), "")
					}
					if globalConfig.OnError == "stop" {
						return errors.WrapError(err, errors.ExecutionError, "SQL执行失败，停止导入")
					}
				} else {
					executedCount++
					if executedCount%100 == 0 {
						globalLogger.Debug("执行进度(非DML阶段)", logger.Int("success", executedCount), logger.Int("errors", errorCount))
					}
				}
			}
			return nil
		})
		if err != nil {
			return err
		}

		// DDL重试
		if len(ddlRetry) > 0 {
			globalLogger.Info("开始对DDL依赖未满足的语句进行重试",
				logger.Int("retry_count", len(ddlRetry)),
				logger.Int("max_retries", globalConfig.MaxRetries),
			)
			for pass := 1; pass <= globalConfig.MaxRetries && len(ddlRetry) > 0; pass++ {
				next := make([]string, 0)
				progressed := 0
				for _, stmt := range ddlRetry {
					up := strings.ToUpper(strings.TrimSpace(stmt))
					if err := exec.ExecuteSingleStatement(stmt); err != nil {
						if isRelationMissing(err) && ((strings.HasPrefix(up, "CREATE TABLE") && strings.Contains(up, "FOREIGN KEY")) || strings.HasPrefix(up, "CREATE VIEW") || strings.HasPrefix(up, "CREATE MATERIALIZED VIEW")) {
							next = append(next, stmt)
							continue
						}
						errorCount++
						globalLogger.Error("SQL执行失败(DDL重试阶段)", logger.String("sql", stmt), logger.ErrorField(err))
						if errorCount >= globalConfig.MaxErrors {
							return errors.NewExecutionError(fmt.Sprintf("错误数量超过限制 (%d)", globalConfig.MaxErrors), "")
						}
						if globalConfig.OnError == "stop" {
							return errors.WrapError(err, errors.ExecutionError, "SQL执行失败，停止导入")
						}
					} else {
						executedCount++
						progressed++
					}
				}
				globalLogger.Info("DDL重试进度", logger.Int("pass", pass), logger.Int("resolved", progressed), logger.Int("remaining", len(next)))
				if progressed == 0 {
					break
				}
				ddlRetry = next
			}
			if len(ddlRetry) > 0 {
				for _, stmt := range ddlRetry {
					errorCount++
					globalLogger.Error("DDL依赖未满足，最终失败", logger.String("sql", stmt))
				}
				if errorCount >= globalConfig.MaxErrors {
					return errors.NewExecutionError(fmt.Sprintf("错误数量超过限制 (%d)", globalConfig.MaxErrors), "")
				}
			}
		}

		// 第二阶段：DML（INSERT/UPDATE/DELETE）
		// 改造为真正流式：并发时通道消费，串行时边解析边执行，避免全量缓冲
		fkRetry := make([]string, 0)

		if globalConfig.EnableParallel {
			queueSize := globalConfig.WorkerCount * globalConfig.ParallelBatchSize
			if queueSize <= 0 {
				queueSize = 1
			}
			globalLogger.Info("使用并发流式执行DML",
				logger.Int("worker_count", globalConfig.WorkerCount),
				logger.Int("queue_size", queueSize),
			)

			dmlChan := make(chan string, queueSize)
			prodErrChan := make(chan error, 1)

			// 生产者：流式解析并把DML语句推入通道
			go func() {
				defer close(dmlChan)
				err := sqlParser.ParseStream(inputFile, func(stmt parser.SQLStatement) error {
					origUp := strings.ToUpper(strings.TrimSpace(stmt.Content))
					if !(strings.HasPrefix(origUp, "INSERT ") || strings.HasPrefix(origUp, "INSERT INTO ") || strings.HasPrefix(origUp, "UPDATE ") || strings.HasPrefix(origUp, "DELETE ")) {
						return nil
					}

					converted, err := converterInstance.Convert(stmt, globalConfig.DataTypeMapping)
					if err != nil {
						return errors.WrapError(err, errors.ConversionError, "SQL语句转换失败")
					}
					trimmed := strings.TrimSpace(converted)
					if trimmed == "" {
						return nil
					}

					// 跳过 psql 元命令（例如 \c / \connect）
					trimNoSpace := strings.TrimSpace(trimmed)
					if strings.HasPrefix(trimNoSpace, "\\c ") || strings.HasPrefix(trimNoSpace, "\\connect ") {
						globalLogger.Debug("跳过psql元命令", logger.String("sql", trimmed))
						return nil
					}

					up := strings.ToUpper(trimmed)
					// 跳过锁表与事务控制
					if strings.HasPrefix(up, "LOCK TABLES") || strings.HasPrefix(up, "UNLOCK TABLES") {
						globalLogger.Debug("跳过MySQL表级锁语句", logger.String("sql", trimmed))
						return nil
					}
					// 跳过MySQL的ALTER TABLE ENABLE/DISABLE KEYS（通常在/*!版本化注释*/中）
					if strings.Contains(up, "ALTER TABLE") && (strings.Contains(up, "ENABLE KEYS") || strings.Contains(up, "DISABLE KEYS")) {
						globalLogger.Debug("跳过MySQL ALTER TABLE KEYS语句", logger.String("sql", trimmed))
						return nil
					}
					if strings.HasPrefix(up, "BEGIN") || strings.HasPrefix(up, "START TRANSACTION") ||
						strings.HasPrefix(up, "COMMIT") || strings.HasPrefix(up, "ROLLBACK") ||
						strings.HasPrefix(up, "SAVEPOINT") || strings.HasPrefix(up, "RELEASE SAVEPOINT") ||
						strings.HasPrefix(up, "SET AUTOCOMMIT") {
						globalLogger.Debug("跳过事务控制语句", logger.String("sql", trimmed))
						return nil
					}

					// 转换后可能包含多语句，按顶层分号拆分并入队
					for _, part := range parser.SplitTopLevelStatements(trimmed) {
						p := strings.TrimSpace(part)
						if p == "" {
							continue
						}
						dmlChan <- p
					}
					return nil
				})
				prodErrChan <- err
			}()

			success, errCnt, fkFailed, execErr := exec.ExecuteStatementsParallelStream(
				dmlChan,
				globalConfig.WorkerCount,
				globalConfig.OnError,
				globalConfig.MaxErrors,
			)

			// 等待生产者结束并检查错误
			prodErr := <-prodErrChan
			if prodErr != nil {
				return prodErr
			}

			executedCount += success
			errorCount += errCnt

			if len(fkFailed) > 0 {
				fkRetry = append(fkRetry, fkFailed...)
				globalLogger.Warn("并发流式执行中检测到外键约束失败，加入重试列表",
					logger.Int("fk_failures", len(fkFailed)),
				)
			}

			if execErr != nil {
				return execErr
			}

			globalLogger.Info("并发流式DML执行完成",
				logger.Int("success", success),
				logger.Int("errors", errCnt),
				logger.Int("fk_retry", len(fkFailed)),
			)
		} else {
			// 串行流式执行：边解析边执行，避免全量缓冲
			err = sqlParser.ParseStream(inputFile, func(stmt parser.SQLStatement) error {
				origUp := strings.ToUpper(strings.TrimSpace(stmt.Content))
				if !(strings.HasPrefix(origUp, "INSERT ") || strings.HasPrefix(origUp, "INSERT INTO ") || strings.HasPrefix(origUp, "UPDATE ") || strings.HasPrefix(origUp, "DELETE ")) {
					return nil
				}

				converted, err := converterInstance.Convert(stmt, globalConfig.DataTypeMapping)
				if err != nil {
					return errors.WrapError(err, errors.ConversionError, "SQL语句转换失败")
				}
				trimmed := strings.TrimSpace(converted)
				if trimmed == "" {
					return nil
				}

				// 跳过 psql 元命令（例如 \c / \connect）
				trimNoSpace := strings.TrimSpace(trimmed)
				if strings.HasPrefix(trimNoSpace, "\\c ") || strings.HasPrefix(trimNoSpace, "\\connect ") {
					globalLogger.Debug("跳过psql元命令", logger.String("sql", trimmed))
					return nil
				}

				up := strings.ToUpper(trimmed)
				// 跳过锁表与事务控制
				if strings.HasPrefix(up, "LOCK TABLES") || strings.HasPrefix(up, "UNLOCK TABLES") {
					globalLogger.Debug("跳过MySQL表级锁语句", logger.String("sql", trimmed))
					return nil
				}
				// 跳过MySQL的ALTER TABLE ENABLE/DISABLE KEYS（通常在/*!版本化注释*/中）
				if strings.Contains(up, "ALTER TABLE") && (strings.Contains(up, "ENABLE KEYS") || strings.Contains(up, "DISABLE KEYS")) {
					globalLogger.Debug("跳过MySQL ALTER TABLE KEYS语句", logger.String("sql", trimmed))
					return nil
				}
				if strings.HasPrefix(up, "BEGIN") || strings.HasPrefix(up, "START TRANSACTION") ||
					strings.HasPrefix(up, "COMMIT") || strings.HasPrefix(up, "ROLLBACK") ||
					strings.HasPrefix(up, "SAVEPOINT") || strings.HasPrefix(up, "RELEASE SAVEPOINT") ||
					strings.HasPrefix(up, "SET AUTOCOMMIT") {
					globalLogger.Debug("跳过事务控制语句", logger.String("sql", trimmed))
					return nil
				}

				for _, part := range parser.SplitTopLevelStatements(trimmed) {
					p := strings.TrimSpace(part)
					if p == "" {
						continue
					}
					if err := exec.ExecuteSingleStatement(p); err != nil {
						if isForeignKeyViolation(err) {
							fkRetry = append(fkRetry, p)
							globalLogger.Warn("检测到外键未满足，延后重试", logger.String("sql", p), logger.ErrorField(err))
							continue
						}
						errorCount++
						globalLogger.Error("SQL执行失败(DML流式串行阶段)",
							logger.String("sql", p),
							logger.ErrorField(err),
							logger.Int("error_count", errorCount),
							logger.Int("max_errors", globalConfig.MaxErrors),
							logger.String("on_error_strategy", globalConfig.OnError),
						)
						if errorCount >= globalConfig.MaxErrors {
							return errors.NewExecutionError(fmt.Sprintf("错误数量超过限制 (%d)", globalConfig.MaxErrors), "")
						}
						if globalConfig.OnError == "stop" {
							return errors.WrapError(err, errors.ExecutionError, "SQL执行失败，停止导入")
						}
					} else {
						executedCount++
						if executedCount%100 == 0 {
							globalLogger.Debug("执行进度(DML流式串行)", logger.Int("success", executedCount), logger.Int("errors", errorCount))
						}
					}
				}
				return nil
			})
			if err != nil {
				return err
			}
		}

		// DML重试
		if len(fkRetry) > 0 {
			globalLogger.Info("开始对外键未满足的语句进行重试",
				logger.Int("retry_count", len(fkRetry)),
				logger.Int("max_retries", globalConfig.MaxRetries),
			)
			for pass := 1; pass <= globalConfig.MaxRetries && len(fkRetry) > 0; pass++ {
				next := make([]string, 0)
				progressed := 0
				for _, stmt := range fkRetry {
					if err := exec.ExecuteSingleStatement(stmt); err != nil {
						if isForeignKeyViolation(err) {
							next = append(next, stmt)
							continue
						}
						errorCount++
						globalLogger.Error("SQL执行失败(重试阶段)", logger.String("sql", stmt), logger.ErrorField(err))
						if errorCount >= globalConfig.MaxErrors {
							return errors.NewExecutionError(fmt.Sprintf("错误数量超过限制 (%d)", globalConfig.MaxErrors), "")
						}
						if globalConfig.OnError == "stop" {
							return errors.WrapError(err, errors.ExecutionError, "SQL执行失败，停止导入")
						}
					} else {
						executedCount++
						progressed++
					}
				}
				globalLogger.Info("外键重试进度", logger.Int("pass", pass), logger.Int("resolved", progressed), logger.Int("remaining", len(next)))
				if progressed == 0 {
					break
				}
				fkRetry = next
			}
			if len(fkRetry) > 0 {
				for _, stmt := range fkRetry {
					errorCount++
					globalLogger.Error("外键未满足，最终失败", logger.String("sql", stmt))
				}
				if errorCount >= globalConfig.MaxErrors {
					return errors.NewExecutionError(fmt.Sprintf("错误数量超过限制 (%d)", globalConfig.MaxErrors), "")
				}
			}
		}

		executeTime := time.Since(startTime)
		globalLogger.Info("导入操作完成流式",
			logger.Int("executed_successfully", executedCount),
			logger.Int("errors", errorCount),
			logger.Duration("execute_time", executeTime),
		)

		if errorCount > 0 {
			globalLogger.Warn("导入完成，但存在错误", logger.Int("error_count", errorCount))
		} else {
			globalLogger.Info("导入成功完成，无错误")
		}

		return nil
	}

	// 解析SQL文件
	globalLogger.Info("解析SQL文件", logger.String("file", inputFile))
	startTime := time.Now()
	statements, err := sqlParser.Parse(inputFile)
	if err != nil {
		return errors.WrapError(err, errors.ParseError, "SQL文件解析失败")
	}
	parseTime := time.Since(startTime)

	globalLogger.Info("SQL文件解析完成",
		logger.Int("statement_count", len(statements)),
		logger.Duration("parse_time", parseTime),
	)

	// 创建转换器
	globalLogger.Debug("创建SQL转换器")
	converterInstance := converter.NewMySQLToKWDBConverter()
	// 将配置中的 max_values_per_insert 传入转换器（顶层与嵌套 conversion 皆可）
	maxValues := globalConfig.MaxValuesPerInsert
	if maxValues == 0 && globalConfig.Conversion.MaxValuesPerInsert != 0 {
		maxValues = globalConfig.Conversion.MaxValuesPerInsert
	}
	if maxValues > 0 {
		converterInstance.SetMaxValuesPerInsert(maxValues)
		globalLogger.Debug("启用INSERT多值拆分", logger.Int("max_values_per_insert", maxValues))
	}

	// 转换SQL语句
	globalLogger.Info("转换SQL语句")
	startTime = time.Now()
	type convertedItem struct {
		sql   string
		typ   parser.StatementType
		line  int
		index int
	}
	convertedItems := make([]convertedItem, 0, len(statements))
	conversionWarnings := make([]string, 0)

	for i, stmt := range statements {
		converted, err := converterInstance.Convert(stmt, globalConfig.DataTypeMapping)
		if err != nil {
			return errors.WrapError(err, errors.ConversionError, "SQL语句转换失败")
		}
		// 对转换结果进行顶层分号拆分，以便在 dry-run 和执行阶段逐条处理
		parts := parser.SplitTopLevelStatements(converted)
		if len(parts) == 0 {
			// 仍然保留空转换（为保持索引对齐），但不追加内容
		} else {
			for _, part := range parts {
				p := strings.TrimSpace(part)
				if p == "" {
					continue
				}
				// 重新识别拆分后片段的语句类型，避免沿用原始类型导致误标
				detectedType := parser.ClassifyStatement(p)
				convertedItems = append(convertedItems, convertedItem{
					sql:   p,
					typ:   detectedType,
					line:  stmt.Line,
					index: i + 1,
				})
			}
		}

		// 获取转换警告
		//warnings := converterInstance.GetConversionWarnings(stmt)
		//conversionWarnings = append(conversionWarnings, warnings...)

		if (i+1)%1000 == 0 {
			globalLogger.Debug("转换进度", logger.Int("converted", i+1), logger.Int("total", len(statements)))
		}
	}
	convertTime := time.Since(startTime)

	globalLogger.Info("SQL语句转换完成",
		logger.Duration("convert_time", convertTime),
		logger.Int("warning_count", len(conversionWarnings)),
	)

	// 记录转换警告
	if len(conversionWarnings) > 0 {
		globalLogger.Warn("转换过程中发现警告", logger.Int("count", len(conversionWarnings)))
		for _, warning := range conversionWarnings {
			globalLogger.Warn("转换警告", logger.String("message", warning))
		}
	}

	// 如果是试运行模式，只输出转换结果
	if globalConfig.DryRun {
		globalLogger.Info("试运行批处理模式，输出转换结果",
			logger.Int("total_statements", len(convertedItems)),
		)
		for i, item := range convertedItems {
			stmtType := parser.GetStatementTypeName(item.typ)
			// 同时输出到控制台和日志
			fmt.Printf("-- Statement %d (orig #%d, %s)\n%s\n\n", i+1, item.index, stmtType, item.sql)
			globalLogger.Info("DRY-RUN转换结果",
				logger.Int("statement_no", i+1),
				logger.Int("orig_statement_no", item.index),
				logger.Int("line", item.line),
				logger.String("type", stmtType),
				logger.String("converted", item.sql),
			)
			if item.index-1 < len(statements) {
				globalLogger.Debug("DRY-RUN原始语句",
					logger.Int("orig_statement_no", item.index),
					logger.String("original", strings.TrimSpace(statements[item.index-1].Content)),
				)
			}
		}
		globalLogger.Info("试运行(批处理)完成",
			logger.Int("converted_count", len(convertedItems)),
		)
		fmt.Printf("\n=== 试运行完成: 共转换 %d 条语句 ===\n", len(convertedItems))
		return nil
	}

	// 创建执行器
	globalLogger.Debug("创建数据库执行器")
	exec, err := executor.NewPgxStatementExecutor(globalConfig.GetConnectionString())
	if err != nil {
		return errors.WrapError(err, errors.ConnectionError, "创建数据库执行器失败")
	}
	defer func() {
		if closeErr := exec.Close(); closeErr != nil {
			globalLogger.Error("关闭数据库连接失败", logger.ErrorField(closeErr))
		}
	}()

	// 测试数据库连接
	globalLogger.Info("测试数据库连接")
	if err := exec.ValidateConnection(); err != nil {
		return errors.WrapError(err, errors.ConnectionError, "数据库连接测试失败")
	}

	// 执行SQL语句
	globalLogger.Info("开始执行SQL语句")
	startTime = time.Now()
	executedCount := 0
	errorCount := 0

	// 第一阶段：执行除DML外的语句（DDL、视图、索引等），避免约束先创建导致后续插入无序失败
	nonDML := make([]string, 0, len(convertedItems))
	dml := make([]string, 0, len(convertedItems))
	for _, item := range convertedItems {
		s := item.sql
		up := strings.ToUpper(strings.TrimSpace(s))
		if strings.HasPrefix(up, "INSERT ") || strings.HasPrefix(up, "INSERT INTO ") ||
			strings.HasPrefix(up, "UPDATE ") || strings.HasPrefix(up, "DELETE ") {
			dml = append(dml, s)
		} else {
			nonDML = append(nonDML, s)
		}
	}

	// 执行非DML语句
	ddlRetry := make([]string, 0)
	for i, stmt := range nonDML {
		trimmed := strings.TrimSpace(stmt)
		if trimmed == "" {
			continue
		}

		// 跳过 psql 元命令（例如 \c / \connect）
		trimNoSpace := strings.TrimSpace(trimmed)
		if strings.HasPrefix(trimNoSpace, "\\c ") || strings.HasPrefix(trimNoSpace, "\\connect ") {
			globalLogger.Debug("跳过psql元命令", logger.String("sql", stmt))
			continue
		}

		// 忽略执行 CREATE DATABASE / DROP DATABASE / USE 语句
		up := strings.ToUpper(trimmed)
		if strings.HasPrefix(up, "CREATE DATABASE") || strings.HasPrefix(up, "DROP DATABASE") || strings.HasPrefix(up, "USE ") {
			globalLogger.Debug("跳过数据库级语句执行", logger.String("sql", stmt))
			continue
		}

		// 忽略MySQL存储过程/函数及DELIMITER相关语句
		if strings.HasPrefix(up, "CREATE PROCEDURE") || strings.HasPrefix(up, "CREATE FUNCTION") ||
			strings.HasPrefix(up, "DROP PROCEDURE") || strings.HasPrefix(up, "DROP FUNCTION") ||
			strings.HasPrefix(up, "DELIMITER ") || strings.HasPrefix(up, "DELIMITER\t") {
			globalLogger.Debug("跳过不兼容的MySQL过程/函数或分隔符语句", logger.String("sql", stmt))
			continue
		}
		// 跳过MySQL存储过程结束标记（各种分隔符形式）：END //、END ;;、END $$、END ;
		reEnd := regexp.MustCompile(`(?i)^\s*END\s*(;{1,2}|//|\$\$)?\s*$`)
		if reEnd.MatchString(trimmed) {
			globalLogger.Debug("跳过MySQL存储过程结束标记", logger.String("sql", stmt))
			continue
		}

		// 跳过MySQL的表级锁语句（KWDB/PG不支持）
		if strings.HasPrefix(up, "LOCK TABLES") || strings.HasPrefix(up, "UNLOCK TABLES") {
			globalLogger.Debug("跳过MySQL表级锁语句", logger.String("sql", stmt))
			continue
		}
		// 跳过MySQL的ALTER TABLE ENABLE/DISABLE KEYS（通常在/*!版本化注释*/中）
		if strings.Contains(up, "ALTER TABLE") && (strings.Contains(up, "ENABLE KEYS") || strings.Contains(up, "DISABLE KEYS")) {
			globalLogger.Debug("跳过MySQL ALTER TABLE KEYS语句", logger.String("sql", stmt))
			continue
		}

		// 跳过事务控制语句（BEGIN/START TRANSACTION/COMMIT/ROLLBACK/SAVEPOINT/RELEASE SAVEPOINT）以及MySQL的SET AUTOCOMMIT
		if strings.HasPrefix(up, "BEGIN") || strings.HasPrefix(up, "START TRANSACTION") ||
			strings.HasPrefix(up, "COMMIT") || strings.HasPrefix(up, "ROLLBACK") ||
			strings.HasPrefix(up, "SAVEPOINT") || strings.HasPrefix(up, "RELEASE SAVEPOINT") ||
			strings.HasPrefix(up, "SET AUTOCOMMIT") {
			globalLogger.Debug("跳过事务控制语句", logger.String("sql", stmt))
			continue
		}

		if err := exec.ExecuteSingleStatement(stmt); err != nil {
			// 依赖的关系不存在(42P01) 时，延后重试：
			// 1) CREATE TABLE (FOREIGN KEY 依赖尚未创建)
			// 2) CREATE VIEW / CREATE MATERIALIZED VIEW (依赖的视图/表尚未创建)
			if isRelationMissing(err) && ((strings.HasPrefix(up, "CREATE TABLE") && strings.Contains(up, "FOREIGN KEY")) || strings.HasPrefix(up, "CREATE VIEW") || strings.HasPrefix(up, "CREATE MATERIALIZED VIEW")) {
				ddlRetry = append(ddlRetry, stmt)
				globalLogger.Warn("检测到DDL依赖未创建，延后重试",
					logger.Int("statement_index", i+1),
					logger.String("sql", stmt),
					logger.ErrorField(err),
				)
				continue
			}
			errorCount++
			globalLogger.Error("SQL执行失败",
				logger.Int("statement_index", i+1),
				logger.String("sql", stmt),
				logger.ErrorField(err),
			)

			// 检查是否超过最大错误数
			if errorCount >= globalConfig.MaxErrors {
				return errors.NewExecutionError(fmt.Sprintf("错误数量超过限制 (%d)", globalConfig.MaxErrors), "")
			}

			// 根据错误处理策略决定是否继续
			if globalConfig.OnError == "stop" {
				return errors.WrapError(err, errors.ExecutionError, "SQL执行失败，停止导入")
			}
		} else {
			executedCount++
		}

		// 显示进度
		if (i+1)%100 == 0 {
			globalLogger.Debug("执行进度(非DML阶段)",
				logger.Int("executed", i+1),
				logger.Int("total", len(nonDML)),
				logger.Int("success", executedCount),
				logger.Int("errors", errorCount),
			)
		}
	}

	// 对延后的DDL进行重试，最多 maxRetries 次，直到不再取得进展
	if len(ddlRetry) > 0 {
		globalLogger.Info("开始对DDL依赖未满足的语句进行重试",
			logger.Int("retry_count", len(ddlRetry)),
			logger.Int("max_retries", globalConfig.MaxRetries),
		)
		for pass := 1; pass <= globalConfig.MaxRetries && len(ddlRetry) > 0; pass++ {
			next := make([]string, 0)
			progressed := 0
			for _, stmt := range ddlRetry {
				up := strings.ToUpper(strings.TrimSpace(stmt))
				if err := exec.ExecuteSingleStatement(stmt); err != nil {
					if isRelationMissing(err) && ((strings.HasPrefix(up, "CREATE TABLE") && strings.Contains(up, "FOREIGN KEY")) || strings.HasPrefix(up, "CREATE VIEW") || strings.HasPrefix(up, "CREATE MATERIALIZED VIEW")) {
						next = append(next, stmt)
						continue
					}
					// 非依赖缺失错误按原策略处理
					errorCount++
					globalLogger.Error("SQL执行失败(DDL重试阶段)",
						logger.String("sql", stmt),
						logger.ErrorField(err),
					)
					if errorCount >= globalConfig.MaxErrors {
						return errors.NewExecutionError(fmt.Sprintf("错误数量超过限制 (%d)", globalConfig.MaxErrors), "")
					}
					if globalConfig.OnError == "stop" {
						return errors.WrapError(err, errors.ExecutionError, "SQL执行失败，停止导入")
					}
				} else {
					executedCount++
					progressed++
				}
			}
			globalLogger.Info("DDL重试进度",
				logger.Int("pass", pass),
				logger.Int("resolved", progressed),
				logger.Int("remaining", len(next)),
			)
			if progressed == 0 {
				// 无进展，提前结束
				break
			}
			ddlRetry = next
		}
		// 仍未解决的视为错误并汇总报告
		if len(ddlRetry) > 0 {
			for _, stmt := range ddlRetry {
				errorCount++
				globalLogger.Error("DDL依赖未满足，最终失败",
					logger.String("sql", stmt),
				)
			}
			if errorCount >= globalConfig.MaxErrors {
				return errors.NewExecutionError(fmt.Sprintf("错误数量超过限制 (%d)", globalConfig.MaxErrors), "")
			}
		}
	}

	// 第二阶段：执行DML语句，遇到外键未满足(23503)的错误时延后重试
	// 先记录DML收集情况，并根据配置选择并发或串行
	globalLogger.Info("DML语句收集完成",
		logger.Int("total_dml", len(dml)),
		logger.Bool("parallel_enabled", globalConfig.EnableParallel),
	)
	fkRetry := make([]string, 0)
	if globalConfig.EnableParallel && len(dml) > 0 {
		// 并发执行
		globalLogger.Info("使用并发模式执行DML语句",
			logger.Int("worker_count", globalConfig.WorkerCount),
			logger.Int("batch_size", globalConfig.ParallelBatchSize),
		)

		success, errors, fkFailed, err := exec.ExecuteStatementsParallel(
			dml,
			globalConfig.WorkerCount,
			globalConfig.ParallelBatchSize,
			globalConfig.OnError,
			globalConfig.MaxErrors,
		)

		executedCount += success
		errorCount += errors

		// 将并发执行中失败的外键语句加入重试列表
		if len(fkFailed) > 0 {
			fkRetry = append(fkRetry, fkFailed...)
			globalLogger.Warn("并发执行中检测到外键约束失败，加入重试列表",
				logger.Int("fk_failures", len(fkFailed)),
			)
		}

		if err != nil {
			return err
		}

		globalLogger.Info("并发DML执行完成",
			logger.Int("success", success),
			logger.Int("errors", errors),
			logger.Int("fk_retry", len(fkFailed)),
		)
	} else {
		for i, stmt := range dml {
			trimmed := strings.TrimSpace(stmt)
			if trimmed == "" {
				continue
			}

			up := strings.ToUpper(trimmed)
			// 跳过明显不执行的语句
			if strings.HasPrefix(up, "LOCK TABLES") || strings.HasPrefix(up, "UNLOCK TABLES") {
				globalLogger.Debug("跳过MySQL表级锁语句", logger.String("sql", stmt))
				continue
			}
			// 跳过MySQL的ALTER TABLE ENABLE/DISABLE KEYS（通常在/*!版本化注释*/中）
			if strings.Contains(up, "ALTER TABLE") && (strings.Contains(up, "ENABLE KEYS") || strings.Contains(up, "DISABLE KEYS")) {
				globalLogger.Debug("跳过MySQL ALTER TABLE KEYS语句", logger.String("sql", stmt))
				continue
			}

			// 跳过事务控制语句（BEGIN/START TRANSACTION/COMMIT/ROLLBACK/SAVEPOINT/RELEASE SAVEPOINT）以及MySQL的SET AUTOCOMMIT
			if strings.HasPrefix(up, "BEGIN") || strings.HasPrefix(up, "START TRANSACTION") ||
				strings.HasPrefix(up, "COMMIT") || strings.HasPrefix(up, "ROLLBACK") ||
				strings.HasPrefix(up, "SAVEPOINT") || strings.HasPrefix(up, "RELEASE SAVEPOINT") ||
				strings.HasPrefix(up, "SET AUTOCOMMIT") {
				globalLogger.Debug("跳过事务控制语句", logger.String("sql", stmt))
				continue
			}

			if err := exec.ExecuteSingleStatement(stmt); err != nil {
				// 外键未满足，延后重试，不计入即时错误数量
				if isForeignKeyViolation(err) {
					fkRetry = append(fkRetry, stmt)
					globalLogger.Warn("检测到外键未满足，延后重试",
						logger.Int("statement_index", i+1),
						logger.String("sql", stmt),
						logger.ErrorField(err),
					)
					continue
				}

				// 其它错误按原策略处理
				errorCount++
				globalLogger.Error("SQL执行失败",
					logger.Int("statement_index", i+1),
					logger.String("sql", stmt),
					logger.ErrorField(err),
				)
				if errorCount >= globalConfig.MaxErrors {
					return errors.NewExecutionError(fmt.Sprintf("错误数量超过限制 (%d)", globalConfig.MaxErrors), "")
				}
				if globalConfig.OnError == "stop" {
					return errors.WrapError(err, errors.ExecutionError, "SQL执行失败，停止导入")
				}
			} else {
				executedCount++
			}

			if (i+1)%100 == 0 {
				globalLogger.Debug("执行进度",
					logger.Int("executed", i+1),
					logger.Int("total", len(dml)),
					logger.Int("success", executedCount),
					logger.Int("errors", errorCount),
				)
			}
		}

	}

	// 对延后的DML执行重试，最多 maxRetries 次，直到不再取得进展
	if len(fkRetry) > 0 {
		globalLogger.Info("开始对外键未满足的语句进行重试",
			logger.Int("retry_count", len(fkRetry)),
			logger.Int("max_retries", globalConfig.MaxRetries),
		)
		for pass := 1; pass <= globalConfig.MaxRetries && len(fkRetry) > 0; pass++ {
			next := make([]string, 0)
			progressed := 0
			for _, stmt := range fkRetry {
				if err := exec.ExecuteSingleStatement(stmt); err != nil {
					if isForeignKeyViolation(err) {
						next = append(next, stmt)
						continue
					}
					// 非外键错误按原策略处理
					errorCount++
					globalLogger.Error("SQL执行失败(重试阶段)",
						logger.String("sql", stmt),
						logger.ErrorField(err),
					)
					if errorCount >= globalConfig.MaxErrors {
						return errors.NewExecutionError(fmt.Sprintf("错误数量超过限制 (%d)", globalConfig.MaxErrors), "")
					}
					if globalConfig.OnError == "stop" {
						return errors.WrapError(err, errors.ExecutionError, "SQL执行失败，停止导入")
					}
				} else {
					executedCount++
					progressed++
				}
			}
			globalLogger.Info("外键重试进度",
				logger.Int("pass", pass),
				logger.Int("resolved", progressed),
				logger.Int("remaining", len(next)),
			)
			if progressed == 0 {
				// 无进展，提前结束
				break
			}
			fkRetry = next
		}
		// 仍未解决的视为错误并汇总报告
		if len(fkRetry) > 0 {
			for _, stmt := range fkRetry {
				errorCount++
				globalLogger.Error("外键未满足，最终失败",
					logger.String("sql", stmt),
				)
			}
			if errorCount >= globalConfig.MaxErrors {
				return errors.NewExecutionError(fmt.Sprintf("错误数量超过限制 (%d)", globalConfig.MaxErrors), "")
			}
		}
	}
	executeTime := time.Since(startTime)

	globalLogger.Info("导入操作完成",
		logger.Int("total_statements", len(convertedItems)),
		logger.Int("executed_successfully", executedCount),
		logger.Int("errors", errorCount),
		logger.Duration("total_time", parseTime+convertTime+executeTime),
		logger.Duration("parse_time", parseTime),
		logger.Duration("convert_time", convertTime),
		logger.Duration("execute_time", executeTime),
	)

	if errorCount > 0 {
		globalLogger.Warn("导入完成，但存在错误", logger.Int("error_count", errorCount))
	} else {
		globalLogger.Info("导入成功完成，无错误")
	}

	return nil
}

// isForeignKeyViolation 判断是否为外键未满足错误(SQLSTATE 23503)
func isForeignKeyViolation(err error) bool {
	if err == nil {
		return false
	}
	msg := strings.ToLower(err.Error())
	return strings.Contains(msg, "sqlstate 23503") || strings.Contains(msg, "violates foreign key constraint")
}

// isRelationMissing 判断是否为关系不存在错误(SQLSTATE 42P01)
func isRelationMissing(err error) bool {
	if err == nil {
		return false
	}
	msg := strings.ToLower(err.Error())
	return strings.Contains(msg, "sqlstate 42p01") || (strings.Contains(msg, "relation") && strings.Contains(msg, "does not exist"))
}

func main() {
	if err := rootCmd.Execute(); err != nil {
		if globalLogger != nil {
			globalLogger.Error("应用程序执行失败", logger.ErrorField(err))
		} else {
			fmt.Fprintf(os.Stderr, "错误: %v\n", err)
		}
		os.Exit(1)
	}
}
